package us.luosl.lslt.asyn;

import us.luosl.lslt.asyn.pipeline.Pipeline;
import us.luosl.lslt.asyn.processor.Processor;
import us.luosl.lslt.asyn.reject.CallerRunsRejectHandler;
import us.luosl.lslt.asyn.reject.RejectHandler;
import us.luosl.lslt.asyn.trigger.Trigger;
import us.luosl.lslt.collection.CollectUtil;

import java.time.LocalDateTime;
import java.util.ArrayList;
import java.util.Collections;
import java.util.List;
import java.util.concurrent.Executors;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.TimeUnit;

/**
 * 异步执行器
 * @param <T> 可以处理的数据类型
 */
public class AsynExecutor<T> {

    /**
     * 默认从 pipeline 取数的间隔
     */
    private static final int DEFAULT_EXECUTE_DURATION = 1000;

    /**
     * 默认的每次处理的数据批次大小
     */
    private static final int DEFAULT_BATCH_SIZE = 100;

    /**
     * 管道
     */
    private Pipeline<T> pipeline;

    /**
     * 触发器
     */
    private Trigger<T> trigger;

    /**
     * 处理器
     */
    private Processor<T> processor;

    /**
     * 拒绝策略
     */
    private RejectHandler<T> rejectHandler;

    /**
     * 定时线程池
     */
    private ScheduledExecutorService scheduledExecutorService = Executors.newScheduledThreadPool(1);

    /**
     * 上次处理时间
     */
    private LocalDateTime lastExecuteTime;

    /**
     * 待处理数据收集器
     */
    private List<T> collector = new ArrayList<>();

    /**
     * 定时执行间隔 毫秒
     */
    private int executeDuration = DEFAULT_EXECUTE_DURATION;

    /**
     * 每次取数大小
     */
    private int batchSize = DEFAULT_BATCH_SIZE;

    public AsynExecutor(Pipeline<T> pipeline, Trigger<T> trigger, Processor<T> processor, RejectHandler<T> rejectHandler) {
        this.pipeline = pipeline;
        this.trigger = trigger;
        this.processor = processor;
        this.rejectHandler = rejectHandler;
    }


    public AsynExecutor(Pipeline<T> pipeline, Trigger<T> trigger, Processor<T> processor) {
        this(pipeline, trigger, processor, new CallerRunsRejectHandler<>());
    }

    /**
     * 启动异步执行器
     */
    public void start() {
        lastExecuteTime = LocalDateTime.now();
        scheduledExecutorService.scheduleWithFixedDelay(this::process,0, executeDuration, TimeUnit.MILLISECONDS);
    }

    /**
     * 压入单个数据
     * @param data 待处理元素
     */
    public void push(T data) {
        push(Collections.singletonList(data));
    }

    /**
     * 批量压入数据
     * @param data 待处理数据集
     */
    public void push(List<T> data){
        List<T> rejectCollector = new ArrayList<>();
        data.forEach(item -> {
            if(!pipeline.push(item)){
                rejectCollector.add(item);
            }
        });
        if(rejectCollector.size() > 0) {
            CollectUtil.partition(data, batchSize).forEach(subData -> rejectHandler.process(processor, subData));
        }
    }

    private void process() {
        try {
            lastExecuteTime = LocalDateTime.now();
            collector.addAll(pipeline.pull(getNeedPullSize()));
            if(trigger.isTrigger(lastExecuteTime, collector)) {
                processor.process(collector);
            }
        }catch (Throwable e){
            start();
            throw e;
        } finally {
            collector = new ArrayList<>();
        }
    }

    /**
     * 获取需要拉取的数据集大小
     * @return 需要拉取的数据集大小
     */
    private int getNeedPullSize(){
        int size = batchSize - collector.size();
        return Math.max(size, 0);
    }

    /**
     * 终止
     */
    public void shutdown() {
        this.scheduledExecutorService.shutdown();
    }

    /**
     * 设置每次执行的批次大小
     * @param batchSize 批次大小
     * @return 异步执行器 AsynExecutor<T>
     */
    public AsynExecutor<T> setBatchSize(int batchSize) {
        if(batchSize <= 0){
            throw new IllegalArgumentException("batchSize must than 0");
        }
        this.batchSize = batchSize;
        return this;
    }

    /**
     * 设置执行间隔
     * @param duration 执行间隔
     * @return 异步执行器 AsynExecutor<T>
     */
    public AsynExecutor<T> setExecuteDuration(int duration) {
        if(duration < 0){
            throw new IllegalArgumentException("execute duration must than or eq 0");
        }
        this.executeDuration = duration;
        return this;
    }
}
